package com.csw.spark

import org.apache.spark.rdd.RDD
import org.apache.spark.storage.StorageLevel
import org.apache.spark.{SparkConf, SparkContext}

object Demo16Cache {
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf()
      .setMaster("local")
      .setAppName("Cache")

    val sc: SparkContext = new SparkContext(conf)

    val studentRDD: RDD[String] = sc.textFile("spark/data/students.txt")

    val mapstudentRDD: RDD[String] = studentRDD.map(i => {
      println("map--")
      i
    })

    /**
      * cache：对多次使用的RDD进行缓存，默认cache是将数据缓存在内存中
      */
    //    mapstudentRDD.cache()


    /**
      * cache：持久化级别选择
      * 1、如果rdd的数量不大，没有超过内存的限制 ---> MEMORY_ONLY
      * 2、如果数据超过了内存的限制  ---> MEMORY_AND_DISK_SER
      * 因为不管压不压缩放在内存都比放在磁盘要快，所以尽量将数据压缩到内存中
      *
      * 缓存实际上是将数据缓存在执行task所在服务器的内存或者磁盘上
      *
      */
    mapstudentRDD.persist(StorageLevel.MEMORY_AND_DISK_SER)

    //统计班级的人数
    val clazzRDD: RDD[(String, Int)] = studentRDD.map(i => (i.split(",")(4), 1))
    val clazzNumRDD: RDD[(String, Int)] = clazzRDD.reduceByKey((x, y) => x + y)

    clazzNumRDD.foreach(println)

    //统计性别的人数
    val genderRDD: RDD[(String, Int)] = studentRDD.map(i => (i.split(",")(3), 1))
    val genderNumRDD: RDD[(String, Int)] = genderRDD.reduceByKey(_ + _)

    genderNumRDD.foreach(println)

  }
}
